package cn.lzd.mr.case9;

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * Created by lzd on 2019/10/22.
 * K数据处理
 * 用未复权的数据替换前复权数据中的成交量和成交额
 */
public class ParserK2 {
    public static class MyMapper extends Mapper<Object, Text, Text, Text> {
        Text outkey = new Text();
        Text outVal = new Text();
        String fileName;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            FileSplit fileSplit = (FileSplit) context.getInputSplit();
            fileName = fileSplit.getPath().getName();
        }

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String str = value.toString();
            String[] arr = str.split(",", -1);
            outkey.set(arr[0]);
            outVal.set(str + "," + fileName);
            context.write(outkey, outVal);
        }
    }

    public static class MyReduce extends Reducer<Text, Text, Text, NullWritable> {
        // 输出类型和Reduce一致
        private MultipleOutputs<Text, NullWritable> mos;
        Text outkey = new Text();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            mos = new MultipleOutputs<>(context);
        }

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            List<String> list = new ArrayList<>();
            Iterator<Text> iterator = values.iterator();
            while (iterator.hasNext()) {
                list.add(iterator.next().toString());
            }
            String a = list.get(0);
            String b = list.get(1);

            String wk;
            String qk;
            if (a.endsWith("wK")) {
                wk = a;
                qk = b;
            } else {
                qk = a;
                wk = b;
            }

            String[] qArr = qk.split(",", -1);
            String[] wArr = wk.split(",", -1);

            String[] arr = qArr[0].split("_", -1);
            KParser kParser = new KParser();
            kParser.setTs_code(arr[0]);
            kParser.setTrade_date(arr[1]);
            kParser.setOpen(qArr[1]);
            kParser.setHigh(qArr[2]);
            kParser.setLow(qArr[3]);
            kParser.setClose(qArr[4]);
            kParser.setVol(wArr[5]);
            kParser.setAmount(wArr[6]);
            outkey.set(JSONObject.toJSONString(kParser));
            mos.write(outkey,NullWritable.get(),kParser.getTs_code());

        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            mos.close();
        }
    }

    public static void parser(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        args = new String[]{"D:\\我的下载\\ticket\\wK,D:\\我的下载\\ticket\\qK", "E:\\ticket_history_data\\parserK"};
        Configuration configuration = new Configuration();

        Job job = Job.getInstance(configuration, "parserK");

        job.setJarByClass(ParserK2.class);

        job.setMapperClass(ParserK2.MyMapper.class);
        job.setReducerClass(ParserK2.MyReduce.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.addInputPaths(job, args[0]);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //加上后不输出part-r-00000文件
        LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }



    public static void main(String[] args) throws Exception {
// parser(args);
    }
}
